package simpledb;

import java.util.*;

/**
 * The Join operator implements the relational join operation.
 * 这里使用的TupleIterator是简单的嵌套循环，把左表和右表每一行数据都进行一次联合查询
 * 优化：github上有将1.左表存入缓存，2.两个表都存入缓存 3.先进行排序再join
 * 暂时不知道为什么要存入缓存，需要查询的表都在缓冲池中，直接可以用，为什么要再用一个结构存储。
 */
public class Join extends Operator {

    private static final long serialVersionUID = 1L;

    private JoinPredicate p;
    private OpIterator child1;
    private OpIterator child2;
    private TupleDesc tupleDesc;
    private TupleIterator nestedLoopTuples;

    static final int BLOCK_SIZE = 4096*32;
    /**
     * Constructor. Accepts two children to join and the predicate to join them
     * on
     * 
     * @param p
     *            The predicate to use to join the children
     * @param child1
     *            Iterator for the left(outer) relation to join
     * @param child2
     *            Iterator for the right(inner) relation to join
     */
    public Join(JoinPredicate p, OpIterator child1, OpIterator child2) {
        // some code goes here
        this.p = p;
        this.child1 = child1;
        this.child2 = child2;
        this.tupleDesc = TupleDesc.merge(child1.getTupleDesc(),child2.getTupleDesc());
    }

    public TupleIterator filter(OpIterator child1, OpIterator child2,JoinPredicate p)throws TransactionAbortedException, DbException {
        List<Tuple> tuples = new ArrayList<>();
        while(child1.hasNext())
        {
            Tuple child1_tuple = child1.next();
            while(child2.hasNext())
            {
                Tuple child2_tuple = child2.next();
                if(p.filter(child1_tuple,child2_tuple))
                {
                    tuples.add(mergeTuple(child1_tuple,child2_tuple));
                }
            }
            child2.rewind();
        }
        return new TupleIterator(tupleDesc,tuples);
    }

    /**
    使用Block Nested-loop Join
    使用一个大小为4096*32byte的缓存，将左表的数据先缓存，减少IO操作
     */
    public TupleIterator blockNestedLoopJoin(OpIterator child1, OpIterator child2,JoinPredicate p) throws TransactionAbortedException, DbException {
        int tuple_num = BLOCK_SIZE/child1.getTupleDesc().getSize();
        Tuple[] cache = new Tuple[tuple_num];
        int index = 0;
        List<Tuple> tuples = new ArrayList<>();
        while(child1.hasNext())
        {
            Tuple tuple1 = child1.next();
            cache[index++] = tuple1;
            if(index>=tuple_num)
            {
                child2.rewind();
                while(child2.hasNext())
                {
                    Tuple tuple2 = child2.next();
                    for(Tuple child1_tuple:cache)
                    {
                        if(p.filter(child1_tuple,tuple2))
                        {
                            tuples.add(mergeTuple(child1_tuple,tuple2));
                        }
                    }
                }
                Arrays.fill(cache,null);
                index = 0;
            }
        }
        if(index>0)
        {
            child2.rewind();
            while(child2.hasNext())
            {
                Tuple child2_tuple = child2.next();
                for(Tuple child1_tuple:cache)
                {
                    if(child1_tuple==null)
                        break;
                    if(p.filter(child1_tuple,child2_tuple))
                    {
                        tuples.add(mergeTuple(child1_tuple,child2_tuple));
                    }
                }
            }
        }
        return new TupleIterator(tupleDesc,tuples);
    }


    public JoinPredicate getJoinPredicate() {
        // some code goes here
        return p;
    }

    /**
     * @return
     *       the field name of join field1. Should be quantified by
     *       alias or table name.
     * */
    public String getJoinField1Name() {
        // some code goes here
        return child1.getTupleDesc().getFieldName(p.getField1());
    }

    /**
     * @return
     *       the field name of join field2. Should be quantified by
     *       alias or table name.
     *       应该用别名或表名来量化
     * */
    public String getJoinField2Name() {
        // some code goes here
        return child2.getTupleDesc().getFieldName(p.getField2());
    }

    /**
     * @see simpledb.TupleDesc#merge(TupleDesc, TupleDesc) for possible
     *      implementation logic.
     */
    public TupleDesc getTupleDesc() {
        // some code goes here
        return tupleDesc;
    }

    public void open() throws DbException, NoSuchElementException,
            TransactionAbortedException {
        // some code goes here
        super.open();
        child1.open();
        child2.open();
        nestedLoopTuples = filter(child1,child2,p);
        nestedLoopTuples.open();
    }

    public void close() {
        // some code goes here
    }

    public void rewind() throws DbException, TransactionAbortedException {
        // some code goes here
        child1.rewind();
        child2.rewind();
        nestedLoopTuples.rewind();
    }



    /**
     * Returns the next tuple generated by the join, or null if there are no
     * more tuples. Logically, this is the next tuple in r1 cross r2 that
     * satisfies the join predicate. There are many possible implementations;
     * the simplest is a nested loops join.
     * <p>
     * Note that the tuples returned from this particular implementation of Join
     * are simply the concatenation of joining tuples from the left and right
     * relation. Therefore, if an equality predicate is used there will be two
     * copies of the join attribute in the results. (Removing such duplicate
     * columns can be done with an additional projection operator if needed.)
     * <p>
     * For example, if one tuple is {1,2,3} and the other tuple is {1,5,6},
     * joined on equality of the first column, then this returns {1,2,3,1,5,6}.
     * 
     * @return The next matching tuple.
     * @see JoinPredicate#filter
     */
    protected Tuple fetchNext() throws TransactionAbortedException, DbException {
        // some code goes here
        if(nestedLoopTuples.hasNext())
        {
          return nestedLoopTuples.next();
        }
        return null;
    }

    public Tuple mergeTuple(Tuple tuple1,Tuple tuple2)
    {
        Tuple merge = new Tuple(getTupleDesc());
        int numFields = merge.getTupleDesc().numFields();
        for (int i = 0; i < tuple1.getTupleDesc().numFields(); i++) {
            merge.setField(i,tuple1.getField(i));
        }
        for (int i = tuple1.getTupleDesc().numFields(); i < numFields; i++) {
            merge.setField(i,tuple2.getField(i-tuple1.getTupleDesc().numFields()));
        }
        return merge;
    }

    @Override
    public OpIterator[] getChildren() {
        // some code goes here
        return new OpIterator[]{child1,child2};
    }

    @Override
    public void setChildren(OpIterator[] children) {
        child1 = children[0];
        child2 = children[1];
        // some code goes here
    }

}
